#coding:utf-8
import sys
from pyspark.conf import SparkConf
from pyspark.context import SparkContext

print(sys.getdefaultencoding())
reload(sys)
sys.setdefaultencoding('utf-8')
print(sys.getdefaultencoding())

def showResult(x):
    print x

#方法
def pv(lines):
    pairSite = lines.map(lambda line:(line.split("\t")[4],1))
    reduceResult = pairSite.reduceByKey(lambda v1,v2:v1+v2)
    result = reduceResult.sortBy(lambda tp:tp[1],ascending=False)
    result.foreach(lambda one:showResult(one))

def uv(lines):
    distinct = lines.map(lambda line:line.split("\t")[1]+"_"+line.split("\t")[4]).distinct()
    reduceResult = distinct.map(lambda distinct:(distinct.split("_")[1],1)).reduceByKey(lambda v1,v2:v1+v2)
    result = reduceResult.sortBy(lambda tp:tp[1],ascending=False)
    result.foreach(lambda one:showResult(one))

def uvExceptBJ(lines,location):
    distinct = lines.filter(lambda line:line.split("\t")[3]<>location).map(lambda line:line.split("\t")[1]+"_"+line.split("\t")[4]).distinct()
    reduceResult = distinct.map(lambda distinct:(distinct.split("_")[1],1)).reduceByKey(lambda v1,v2:v1+v2)
    result = reduceResult.sortBy(lambda tp:tp[1],ascending=False)
    result.foreach(lambda one:showResult(one))

def getCurrSiteTop2Location(one):
    site = one[0]
    locations = one[1]
    locationDict = {}
    for location in locations:
        if location in locationDict:
            locationDict[location] += 1
        else:
            locationDict[location] = 1
    resultList = [] 
    sortedList = sorted(locationDict.items(),key = lambda kv:kv[1],reverse=True)        
    if len(sortedList)<2:
        resultList = sortedList
    else:
        for i in range(2):
            resultList.append(sortedList[i])
    return site,resultList
    
def getTop2Location(lines):
    site_locations = lines.map(lambda line:(line.split("\t")[4],line.split("\t")[3])).groupByKey()
    result = site_locations.map(lambda one:getCurrSiteTop2Location(one)).collect()
    for elem in result:
        print(elem)
       
def getSiteInfo(one):
    userid = one[0]
    sites = one[1]
    dic = {}
    for site in sites:
        if site in dic:
            dic[site]+=1
        else:
            dic[site] = 1
    resultList = []
    for site,count in dic.items():
        resultList.append((site,(userid,count)))
    return resultList
 
def getCurSiteTop3User(one):
    site = one[0]
    userid_count_Iterable = one[1]
    top3List = ["","",""]
    for userid_count in userid_count_Iterable:
        for i in range(0,len(top3List)):
            if top3List[i] == "":
                top3List[i] = userid_count
                break
            else:
                if userid_count[1]>top3List[i][1]:
                    for j in range(2,i,-1):
                        top3List[j] = top3List[j-1]
                    top3List[i] = userid_count
                break        
    return site,top3List
    
    
def getTop3User(lines):
    site_uid_count = lines.map(lambda line:(line.split("\t")[2],line.split("\t")[4])).groupByKey().flatMap(lambda one:getSiteInfo(one))
    result = site_uid_count.groupByKey().map(lambda one:getCurSiteTop3User(one)).collect()
    for elem in result:
        print(elem)
    
    
if __name__=='__main__':
    conf = SparkConf().setMaster("local").setAppName("test")
    sc = SparkContext(conf=conf)
    lines = sc.textFile("/Users/mac/Downloads/workspace/kkb-bigdata/src/main/python/com/gy/pvuv/pvuvdata")
    # pv(lines)
    # uv(lines)
    #某个地区以外的uv
    # uvExceptBJ(lines,'beijing')

    # 统计每个网站最活跃的top2地区
    # getTop2Location(lines)

    # 统计每个网站下最活跃的top3用户
    getTop3User(lines)
    
    
    
    
    
    
    
    
    